package com.talent.core.protocol.mqtt.handler;

import com.talent.core.protocol.mqtt.TlMqttSession;
import com.talent.core.protocol.mqtt.config.Constant;
import com.talent.core.protocol.mqtt.model.request.TlMqttPubAckReq;
import com.talent.core.protocol.mqtt.model.request.TlMqttPublishReq;
import com.talent.core.protocol.mqtt.model.variable.TlMqttPubAckVariableHead;
import com.talent.core.protocol.mqtt.model.variable.TlMqttPublishVariableHead;
import com.talent.core.protocol.mqtt.store.IMqttStore;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Set;

/**
 * @ClassName：ConnectHandler
 * @Author: hszhou
 * @Date: 2024/11/25 11:06
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
public class TlMqttPubAckHandler extends SimpleChannelInboundHandler<TlMqttPubAckReq> {

    private static final  Logger logger = LoggerFactory.getLogger(TlMqttPubAckHandler.class);

    private IMqttStore mqttStore;

    public TlMqttPubAckHandler(IMqttStore mqttStore){
        this.mqttStore = mqttStore;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttPubAckReq msg) throws Exception {

        logger.debug("进入到【PUBACK】的处理器");
        Object clientId = ctx.channel().attr(AttributeKey.valueOf(Constant.CLIENT_ID)).get();
        if(clientId!=null){
            TlMqttPubAckVariableHead variableHead = msg.getVariableHead();
            Integer messageId = variableHead.getMessageId();
            logger.info("收到客户端{}的消息{}确定",clientId,messageId);
            //获取到会话 如果有会话 那么就将qos1的消息拿出来 这里确认后就移除
            TlMqttSession session = mqttStore.getSessionByClientId(clientId.toString());
            //删除本地存储的消息
            mqttStore.removeStoreMsg(messageId);

            //如果会话不是空 且是保留会话
            if(session!=null && session.getCleanSession()==0){
                Set<TlMqttPublishReq> republish = session.getRepublish();
                Iterator<TlMqttPublishReq> iterator = republish.iterator();
                while (iterator.hasNext()){
                    TlMqttPublishReq next = iterator.next();
                    TlMqttPublishVariableHead vh = next.getVariableHead();
                    if(vh.getMessageId().toString().equals(messageId.toString())){
                        logger.info("向在线的客户端{}保存会话移除的消息{}",session.getClientId(),
                                messageId);
                        iterator.remove();
                    }
                }
            }
        }

    }
}
